package org.rency.mq.activemq.jms.service.impl;

import org.rency.mq.activemq.constant.AcknowledgeMode;
import org.rency.mq.activemq.constant.DeliveryMode;
import org.rency.mq.activemq.constant.DestinationType;
import org.rency.mq.activemq.domain.DestinationInfo;
import org.rency.mq.activemq.exception.MQException;
import org.rency.mq.activemq.jms.factory.DestinationInfoFactory;
import org.rency.mq.activemq.jms.service.JmsAccessor;
import org.rency.mq.activemq.jms.support.JmsTemplate;
import org.rency.mq.activemq.request.MQRequest;
import org.rency.mq.activemq.utils.MessageConvert;
import org.springframework.jms.JmsException;
import org.springframework.jms.core.MessageCreator;
import org.springframework.util.Assert;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import java.util.Map;
import java.util.Set;

public class JmsAccessorImpl implements JmsAccessor {

	private JmsTemplate jmsTemplate;
	private DestinationInfoFactory destinationInfoFactory;

	public void sendMessage(final MQRequest request) throws MQException {
		DestinationInfo destinationInfo = resolveDestination(request.getActionId());
		request.setDestination(destinationInfo.getDestination());
		request.setDestinationType(destinationInfo.getDestinationType());
		try {
			sendMessage(request.getDestination(), request.getDestinationType(),request);
		} catch (JMSException e) {
			throw new MQException("Send message failed.", e);
		}
	}

	public void sendMessage(String destination, int destinationType, MQRequest request) throws JMSException {
		sendMessage(destination, destinationType, request.getMessageFormat(), request.isTransacted(), request.getAcknowledgeMode(), request.getDeliveryMode(), request);
	}

	public void sendMessage(String destination, int destinationType, final int messageFormat, boolean transacted, int acknowledgeMode,
			final int deliveryMode, final Object request) throws JMSException {
		jmsTemplate.setDeliveryPersistent(deliveryMode == DeliveryMode.PERSISTENT);
		jmsTemplate.setSessionTransacted(transacted);
		jmsTemplate.setSessionAcknowledgeMode(acknowledgeMode);
		jmsTemplate.setPubSubDomain(destinationType == DestinationType.TOPIC);
		jmsTemplate.send(destination,destinationType,transacted,acknowledgeMode,deliveryMode, new MessageCreator() {
					public Message createMessage(Session session)throws JMSException {
						Message message = MessageConvert.toMessage(session, request, messageFormat);
						message.setJMSDeliveryMode(deliveryMode);
						if(request instanceof MQRequest){
							setMessageProperties(message, ((MQRequest)request).getProperties());
						}
						return message;
					}
				});
	}

	public Object receiveMessage(int actionId) throws MQException {
		return receiveMessage(actionId, false, AcknowledgeMode.DUPS_OK_ACKNOWLEDGE, null);
	}

	public Object receiveMessage(int actionId, boolean transacted, int acknowledgeMode, String selector) throws MQException {
		Assert.notNull(actionId, "MQRequest action id cannot be null.");
		Assert.isTrue(actionId > 0, "MQRequest action id must be greater than zero.");
		DestinationInfo destinationInfo = resolveDestination(actionId);
		return receiveMessage(destinationInfo.getDestination(), destinationInfo.getDestinationType(), transacted, acknowledgeMode, selector);
	}
	
	public Object receiveMessage(String queueName) throws MQException {
		return receiveMessage(queueName, DestinationType.QUEUE);
	}

	public Object receiveMessage(String destination, int destinationType) throws MQException {
		return receiveMessage(destination, destinationType, null);
	}

	public Object receiveMessage(String destination, int destinationType, String selector) throws MQException {
		return receiveMessage(destination, destinationType, false, AcknowledgeMode.AUTO_ACKNOWLEDGE, selector);
	}

	public Object receiveMessage(String destination, int destinationType, boolean transacted, 
			int acknowledgeMode, String selector) throws MQException {		
		try {
			jmsTemplate.setSessionTransacted(transacted);
			jmsTemplate.setSessionAcknowledgeMode(acknowledgeMode);
			jmsTemplate.setPubSubDomain(destinationType == DestinationType.TOPIC);
			Message message = jmsTemplate.receiveSelected(destination, selector);
			if (message == null) {
				return null;
			}
			return MessageConvert.fromMessage(message);
		} catch (JmsException e) {
			throw new MQException("Receive message failed.", e);
		} catch (JMSException e) {
			throw new MQException("Message received is invalid.", e);
		}
	}

	private void setMessageProperties(Message message, Map<String, Object> props) throws JMSException {
		if (props == null || props.isEmpty()) {
			return;
		}
		Set<String> keys = props.keySet();

		for (String key : keys) {
			Object value = props.get(key);

			if (value instanceof String) {
				message.setStringProperty(key, (String) value);
			} else if (value instanceof Integer) {
				message.setIntProperty(key, (Integer) value);
			} else if (value instanceof Boolean) {
				message.setBooleanProperty(key, (Boolean) value);
			} else if (value instanceof Long) {
				message.setLongProperty(key, (Long) value);
			} else if (value instanceof Double) {
				message.setDoubleProperty(key, (Double) value);
			} else if (value instanceof Float) {
				message.setFloatProperty(key, (Float) value);
			} else {
				message.setObjectProperty(key, value);
			}
		}
	}

	private DestinationInfo resolveDestination(Integer actionId) {
		Assert.notNull(actionId, "MQRequest action id cannot be null.");
		Assert.isTrue(actionId > 0,"MQRequest action id must be greater than zero.");
		DestinationInfo destinationInfo = destinationInfoFactory.getDestinationInfo(actionId);
		if (destinationInfo == null) {
			throw new IllegalArgumentException("Invalid actionId.");
		}
		return destinationInfo;
	}

	public void setJmsTemplate(JmsTemplate jmsTemplate) {
		this.jmsTemplate = jmsTemplate;
	}

	public void setDestinationInfoFactory(DestinationInfoFactory destinationInfoFactory) {
		this.destinationInfoFactory = destinationInfoFactory;
	}

}
